package cn.antcore.kafka.core;

import cn.antcore.kafka.ConsumerCallback;
import cn.antcore.kafka.SendCallback;
import cn.antcore.kafka.config.KafkaConfig;

/**
 * AntKafka
 * Created By Hong on 2018/8/22
 **/
public class AntKafka<K, V> {


    /**
     * 创建AntKafka
     *
     * @param topic topic
     * @param <K>   泛型K
     * @param <V>   泛型V
     * @return AntKafka
     */
    public static <K, V> AntKafka<K, V> create(String topic) {
        return new AntKafka<>(topic);
    }

    /**
     * 创建AntKafka
     *
     * @param topic            topic
     * @param bootstrapService Kafka服务器地址
     * @param <K>              泛型K
     * @param <V>              泛型V
     * @return AntKafka
     */
    public static <K, V> AntKafka<K, V> create(String topic, String bootstrapService) {
        return new AntKafka<>(topic, bootstrapService);
    }

    /**
     * 创建AntKafka
     *
     * @param topic  topic
     * @param config 配置
     * @param <K>    泛型K
     * @param <V>    泛型V
     * @return AntKafka
     */
    public static <K, V> AntKafka<K, V> create(String topic, KafkaConfig config) {
        return new AntKafka<>(topic);
    }

    /**
     * 创建AntKafka
     *
     * @param topic            topic
     * @param bootstrapService Kafka服务器地址
     * @param config           配置
     * @param <K>              泛型K
     * @param <V>              泛型V
     * @return AntKafka
     */
    public static <K, V> AntKafka<K, V> create(String topic, String bootstrapService, KafkaConfig config) {
        return new AntKafka<>(topic, bootstrapService);
    }

    private final String topic;
    private final KafkaConfig config;

    /**
     * 构造函数
     *
     * @param topic topic
     */
    private AntKafka(String topic) {
        this.topic = topic;
        this.config = new KafkaConfig();
    }

    /**
     * 构造函数
     *
     * @param topic            topic
     * @param bootstrapService bootstrapService
     */
    private AntKafka(String topic, String bootstrapService) {
        this.topic = topic;
        this.config = new KafkaConfig(bootstrapService);
    }

    /**
     * 构造函数
     *
     * @param topic  topic
     * @param config config
     */
    private AntKafka(String topic, KafkaConfig config) {
        this.topic = topic;
        this.config = config;
    }

    /**
     * 发送消息
     *
     * @param v Value
     */
    public void send(V v) {
        AntProducer.get(topic, this.config.getProducerConfig())
                .send(v, new SendCallback.Empty());
    }

    /**
     * 发送消息
     *
     * @param v        Value
     * @param callback 回调
     */
    public void send(V v, SendCallback callback) {
        AntProducer.get(topic, this.config.getProducerConfig())
                .send(v, callback);
    }

    /**
     * 发送消息
     *
     * @param k Key
     * @param v Value
     */
    public void send(K k, V v) {
        AntProducer.get(topic, this.config.getProducerConfig())
                .send(k, v, new SendCallback.Empty());
    }

    /**
     * 发送消息
     *
     * @param k        Key
     * @param v        Value
     * @param callback 回调
     */
    public void send(K k, V v, SendCallback callback) {
        AntProducer.get(topic, this.config.getProducerConfig())
                .send(k, v, callback);
    }

    /**
     * 接收消息
     *
     * @param groupId  groupId
     * @param callback 回调
     */
    public void receive(String groupId, ConsumerCallback<K, V> callback) {
        new AntConsumer<>(topic, groupId, this.config.getConsumerConfig())
                .addCallback(callback);
    }
}
